package com.axend.aerosense.common.connect.tcp;

import com.axend.aerosense.base.utils.ByteUtils;
import com.axend.aerosense.base.utils.LogUtils;
import com.axend.aerosense.base.utils.Values;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

public class TcpClient {

    //设备服务器地址，端口
    public final static String WIFI_SERVER_IP = "10.10.100.254";
    public final static int WIFI_SERVER_PORT = 8899;

    //连接状态
    public final static int STATE_FAILED = 100;  //连接失败
    public final static int STATE_BROKEN = 200;  //连接中断
    public final static int STATE_DISCONNECT = 300;  //未连接
    public final static int STATE_SEND_TIMEOUT = 400;  //消息已经发送，且在规定时间内未回消息，则为发送超时
    public final static int STATE_SEND_SUCCESS = 500;  //消息发送成功，并接收到返回数据
    public final static int STATE_SENDING = 600;  //消息正在发送中
    public final static int STATE_SEND_MESSAGE_FAILED = 700;  //消息发送失败
    //控制客户端监听线程的结束
    private boolean run = true;
    //输入输出流
    private InputStream inputStream;
    private OutputStream outputStream;
    private Socket socket;
    private static TcpClient tcpClient;
    private boolean isConnected = false;
    private Callback mCallback;

    private static String mTcpIp = WIFI_SERVER_IP;
    private static int mTcpPort = WIFI_SERVER_PORT;

    private boolean isSendThreadRun = true;  //发送消息线程是否运行
    private boolean isSend = false;  //是否立马发送消息
    private byte[] sendData;  //发送的数据
    private int createConnectCount = 3;  //创建连接次数

    private static Observable<String> data;
    private static Observable<Long> time;
    static {
        //间隔发送时间2秒，最多重发3次
        List<String> list = new ArrayList<>();
        for (int i = 1; i <= 3; i++){
            list.add(i + "");
        }
        data = Observable.fromIterable(list);
        time = Observable.interval(2000, TimeUnit.MILLISECONDS);
    }

    private Disposable disposable;

    /**
     * 私有构造，外部无法直接创建
     */
    private TcpClient(){
        initSendThread();
    }

    /**
     * 单例模式
     * @return
     */
    public static TcpClient getInstance(){
        synchronized (TcpClient.class){
            if (tcpClient == null){
                tcpClient = new TcpClient();
            }
            return tcpClient;
        }
    }

    //自定义ip、端口
    public static TcpClient getInstance(String tcpIp, int tcpPort){
        if (tcpClient == null){
            tcpClient = new TcpClient();
        }
        mTcpIp = tcpIp;
        mTcpPort = tcpPort;
        return tcpClient;
    }

    //设置回调
    public void setCallback(Callback callback){
        this.mCallback = callback;
    }

    /**
     * 创建连接
     * @throws IOException
     */
    public void createConnect(){
        if (!isConnected){
            LogUtils.d(this, "创建连接.......");
            new Thread(){
                @Override
                public void run() {
                    //重连,间隔1秒，否则连接失败！
                    int i = 1;
                    do{
                        try {
                            sleep(1000);
                            socket = new Socket(mTcpIp, mTcpPort);
                            new ClientThread(socket).start();
                            break;
                        }catch (Exception e){
                            e.printStackTrace();
                            LogUtils.e(this, "创建连接失败！");
                            if(null != socket){
                                try {
                                    socket.close();
                                } catch (IOException ioException) {
                                    ioException.printStackTrace();
                                }
                            }

                        }
                        i++;

                    }while (i < createConnectCount);

                    if (i == createConnectCount){
                        callbackFailed(STATE_FAILED);
                    }

                }
            }.start();
        }else {
            callbackConnected();
        }
    }

    private ClientThread clientThread;

    /** 在当前线程创建连接 */
    public Socket createConnectByCurrentThread(){
        if (!isConnected){
            LogUtils.d(this, "创建连接.......");
            try {
                if(null != socket){
                    socket.close();
                }
                socket = new Socket(mTcpIp, mTcpPort);
                clientThread = new ClientThread(socket);
                clientThread.start();
            }catch (Exception e){
                e.printStackTrace();
                LogUtils.e(this, "创建连接失败！");
                callbackFailed(STATE_FAILED);
            }
        }else {
            callbackConnected();
        }
        return socket;
    }

    /**
     * 发送数据（新线程发送数据）
     * @param data
     */
    public void writeData(final byte[] tag, final byte[] data) {
        if (isConnected && data.length > 0){
            new Thread(){
                @Override
                public void run() {
                    try {
                        if (outputStream != null) {
                            byte[] head = Values.HEAD_BYTE;
                            byte[] lengthByte = ByteUtils.intToBytesHigh2Low(data.length);
                            byte[] length = {lengthByte[0], lengthByte[1]};
                            byte[] msg = ByteUtils.inversionArr(data);
                            List<byte[]> list = new ArrayList<>();
                            list.add(head);
                            list.add(length);
                            list.add(tag);
                            list.add(msg);
                            byte[] result = ByteUtils.mergeByteArray(list);
                            LogUtils.d(this, "发送数据数据：" + ByteUtils.bytesToHexString(result));
                            outputStream.write(result);
                        }
                    }catch (IOException e){
                        callbackFailed(STATE_SEND_MESSAGE_FAILED);
                        e.printStackTrace();
                        LogUtils.e(this, "写入数据失败！");
                    }
                }
            }.run();
        }else {
            LogUtils.e(this, "数据发送失败，服务器未连接!");
            callbackFailed(STATE_DISCONNECT);
        }
    }

    /**
     * 发送数据（专门发送数据线程、添加消息重发机制）
     * @param data
     */
    public void writeData(byte[] data) {
        if (data.length > 0){
            this.sendData = data;
            //发送
            isSend = true;
            msgResend(data);
        }
    }

    /**
     * 发送数据（固件升级专用）
     * @param data
     */
    public void updateWriteData(byte[] data) {
        if (data.length > 0){
            this.sendData = data;
            //发送
            isSend = true;
        }
    }


    /**
     * 关闭连接
     */
    public void closeConnect(){
        if (isConnected){
            try {
                setRun(false);
                socket.shutdownOutput();
                socket.shutdownInput();
                tcpClient = null;
                //关闭发送线程
                isSendThreadRun = false;
//                if (inputStream != null){
//                    inputStream.close();
//                }
//                if (outputStream != null){
//                    outputStream.close();
//                }
//                mCallback = null;
            }catch (IOException e){
                e.printStackTrace();
            }
            isConnected = false;
        }else {
            LogUtils.e(this, "服务器已关闭，请勿重复操作！");
        }
    }

    /**
     * 是否连接
     * @return
     */
    public boolean isConnected() {
        return isConnected;
    }

    /**
     * 监听消息发送状态，实现消息重新发送（不完善）
     */
    public void msgResend(final byte[] msg){
        disposable = Observable.zip(data, time, new BiFunction<String, Long, Object>() {
            @Override
            public Object apply(String s, Long aLong){
                try {
                    if (isConnected){
                        outputStream.write(msg);
                        LogUtils.d(TcpClient.this, "重发消息第" + s + "次：" + ByteUtils.bytesToHexString(msg));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return s;
            }
        })
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<Object>() {
            @Override
            public void accept(Object o) throws Exception {
            }
        });
    }

    private Thread sendThread;

    //初始化发送数据线程
    private void initSendThread(){
        sendThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (isSendThreadRun){
                    if(isSend){
                        if (isConnected) {
                            try {
                                LogUtils.d(this, "发送数据数据：" + ByteUtils.bytesToHexString(sendData));
                                outputStream.write(sendData);
                            } catch (IOException e) {
                                callbackFailed(STATE_SEND_MESSAGE_FAILED);
                                LogUtils.e(this, "写入数据失败！");
                                e.printStackTrace();
                            }
                        }else {
                            if (disposable != null){
                                disposable.dispose();
                            }
                            LogUtils.e(this, "数据发送失败，服务器未连接!");
                            callbackFailed(STATE_DISCONNECT);
                        }
                        isSend = false;
                    }
                }
            }
        });
        sendThread.start();
    }

    private class ClientThread extends Thread {
        private Socket cSocket;

        public ClientThread(Socket s) {
            this.cSocket = s;
        }

        @Override
        public void run() {
            try {
                //获取数据流
                inputStream = cSocket.getInputStream();
                outputStream = cSocket.getOutputStream();
                byte[] buffer = new byte[128];
                int bytes = 0;
                LogUtils.d(this, "连接成功，正在接收数据。。。");
                isConnected = true;
                callbackConnected();
                while (isRun()) {
                    //读取数据
                    bytes = inputStream.read(buffer);
                    //接收到消息，取消重发
                    if (disposable != null){
                        disposable.dispose();
                    }
                    if (bytes > 0){
                        byte[] data = new byte[bytes];
                        //将字节数组复制到新长度数组
                        System.arraycopy(buffer, 0, data, 0, bytes);
                        LogUtils.d(TcpClient.this, "接收到的原始数据:" + ByteUtils.bytesToHexString(data));
                        //回调数据
                        //判断是雷达设置的命令还是升级的命令
                        String hexString = ByteUtils.bytesToHexString(data);
//                        if (hexString.indexOf(Values.HEAD) != -1 && hexString.length() < 40 || data[4] == 0x02){
//
//                        } else {
//                            mCallback.readData(Values.UPDATE_COMMAND, data);
//                        }
                        //此处可能会收到非AAAA5555头的字节数据，这里丢掉不处理
                        if (hexString.contains(Values.HEAD) && hexString.length() > 8){
                            //获取标识符
                            String tag = ByteUtils.bytesToHexString(new byte[]{data[5], data[4]});
                            //此处获取雷达id不需要转换
                            if (data[4] != 0x02){
                                //大小端转换
                                data = ByteUtils.toLowcase(data);
                            }
                            LogUtils.d(TcpClient.this,"-------->接收到的数据转换后：" + ByteUtils.bytesToHexString(data));
                            byte[] temp = new byte[data.length - 8];
                            System.arraycopy(data, 8, temp, 0, temp.length);
                            callbackReadData(tag, temp);
                        }
                    }
                }
                isConnected = false;
                socket.close();
                inputStream.close();
                outputStream.close();
                LogUtils.d(this, "连接已断开");
            }catch(IOException e){
                e.printStackTrace();
                LogUtils.e(this, "连接中断");
                isConnected = false;
                callbackFailed(STATE_BROKEN);
                closeConnect();
                try {
                    isConnected = false;
                    socket.close();
                    inputStream.close();
                    outputStream.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }

    /**
     * 获取到数据回调接口
     */
    public interface Callback{
        //已经成功连接后回调
        void connected();
        //连接失败回调
        void failed(int state);
        //获取到数据后的回调
        void readData(String tag, byte[] data);
    }

    private void callbackConnected(){
        if (null != mCallback){
            mCallback.connected();
        }else {
            LogUtils.d(this, "没有设置回调！");
        }
    }

    private void callbackFailed(int state){
        if (null != mCallback){
            mCallback.failed(state);
        }else {
            LogUtils.d(this, "没有设置回调！");
        }
    }

    private void callbackReadData(String tag, byte[] data){
        if (null != mCallback){
            mCallback.readData(tag, data);
        }else {
            LogUtils.d(this, "没有设置回调！");
        }
    }

    private synchronized boolean isRun() {
        return run;
    }

    private void setRun(boolean run) {
        this.run = run;
    }

    @Override
    protected void finalize() throws Throwable {
        if(null != inputStream){
            inputStream.close();
            inputStream = null;
        }
        if(outputStream != null){
            outputStream.close();
            outputStream = null;
        }

        if(null != clientThread && !clientThread.isInterrupted()){
            clientThread.interrupt();
            clientThread = null;
        }
        if(null != sendThread && !sendThread.isInterrupted()){
            sendThread.interrupt();
            sendThread = null;
        }
        if(null != disposable&& !disposable.isDisposed()){
            disposable.dispose();
            disposable = null;
        }
        if(null != socket){
            socket.shutdownOutput();
            socket.shutdownInput();
            socket.close();
            socket = null;
        }

    }
}
